package cloud.xiguapi.ubas.analysis.advertising;

import cloud.xiguapi.ubas.analysis.advertising.model.*;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.TimeCharacteristic;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;
import org.apache.flink.streaming.api.functions.timestamps.AscendingTimestampExtractor;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.util.OutputTag;
import org.springframework.scheduling.annotation.Async;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;

import static cloud.xiguapi.ubas.constant.JobNameConstant.ADVERTISING;

/**
 * @author 大大大西西瓜皮🍉
 * date: 2021-5-19 下午 05:58
 * desc:
 */
@Component
public class BlacklistProvinceAdvertisingAnalysis {

    @Resource(name = "env")
    private StreamExecutionEnvironment env;

    @Async("blacklistProvinceAdvertisingAnalysis")
    public void analysis(String path) throws Exception {
        env.setParallelism(8);
        env.setStreamTimeCharacteristic(TimeCharacteristic.EventTime);

        // 读取数据
        DataStreamSource<String> inputDataStream = env.readTextFile(path);

        // 转换操作
        SingleOutputStreamOperator<ProvinceAdvertisingClickEvent> advertisingDataStream = inputDataStream.map(line -> {
            String[] fields = line.split(",");
            return ProvinceAdvertisingClickEvent.builder()
                    .userId(Long.valueOf(fields[0]))
                    .advertisingId(Long.valueOf(fields[1]))
                    .province(fields[2])
                    .city(fields[3])
                    .timestamp(Long.valueOf(fields[4]))
                    .build();
        }).assignTimestampsAndWatermarks(new AscendingTimestampExtractor<ProvinceAdvertisingClickEvent>() {
            @Override
            public long extractAscendingTimestamp(ProvinceAdvertisingClickEvent element) {
                return element.getTimestamp() * 1000L;
            }
        });

        // 针对同一个用户点击同一个广告的行为进行检测报警
        SingleOutputStreamOperator<ProvinceAdvertisingClickEvent> filterAdvertisingDataStream = advertisingDataStream
                // 基于用户ID和广告ID分组
                .keyBy("userId", "advertisingId")
                .process(new BlacklistUserFilter(100));

        // 打印报警的侧输出流信息
        filterAdvertisingDataStream.getSideOutput(new OutputTag<BlacklistUserWarning>("blacklist") {
        }).print();

        // 对过滤后的数据开窗聚合
        SingleOutputStreamOperator<ProvinceAdvertisingCountView> adResultDataStream = filterAdvertisingDataStream
                .keyBy(ProvinceAdvertisingClickEvent::getProvince)
                // .keyBy("province", "userId")
                .timeWindow(Time.hours(1L), Time.minutes(5L))
                .aggregate(new ProvinceAdvertisingCountAggregate(), new ProvinceAdvertisingCountResult());

        adResultDataStream.print();
        adResultDataStream.addSink(new BlacklistProvinceAdvertisingMySqlSink());

        env.execute(ADVERTISING.getJobName());
    }

    public static class BlacklistProvinceAdvertisingMySqlSink extends RichSinkFunction<ProvinceAdvertisingCountView> {

        private transient Connection connection;

        private transient PreparedStatement clearStatement;

        private transient PreparedStatement selectStatement;

        private transient PreparedStatement insertStatement;

        private transient PreparedStatement updateStatement;

        private transient boolean isFirst = true;

        @Override
        public void open(Configuration parameters) throws Exception {
            connection = DriverManager.getConnection("jdbc:mysql://localhost:3306/ubas?serverTimezone=UTC", "root", "1015");
            clearStatement = connection.prepareStatement("TRUNCATE TABLE blacklist_advertising");
            selectStatement = connection.prepareStatement("SELECT 1 FROM blacklist_advertising WHERE province = ? LIMIT 1");
            insertStatement = connection.prepareStatement("INSERT INTO blacklist_advertising(province, window_end, cnt) VALUES(?, ?, ?)");
            updateStatement = connection.prepareStatement("UPDATE blacklist_advertising SET cnt = ?, window_end = ? WHERE province = ?");
        }

        @Override
        public void close() throws Exception {
            updateStatement.close();
            selectStatement.close();
            clearStatement.close();
            insertStatement.close();
            connection.close();
        }

        @Override
        public void invoke(ProvinceAdvertisingCountView value, SinkFunction.Context context) throws Exception {
            if (isFirst) {
                clearStatement.execute();
                isFirst = false;
                return;
            }

            selectStatement.setString(1, value.getProvince());
            boolean next = selectStatement.executeQuery().next();

            if (next) {
                updateStatement.setLong(1, value.getCount());
                updateStatement.setString(2, value.getWindowEnd());
                updateStatement.setString(3, value.getProvince());
                updateStatement.execute();
            } else {
                insertStatement.setString(1, value.getProvince());
                insertStatement.setString(2, value.getWindowEnd());
                insertStatement.setLong(3, value.getCount());
                insertStatement.execute();
            }
        }
    }
}
